package doris;


import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.connector.jdbc.internal.converter.JdbcRowConverter;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.types.Row;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;

public class Demo1 {

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String jdbcUrl = "jdbc:mysql://sjz02:9030/demo";
        String jdbcDriver = "com.mysql.cj.jdbc.Driver";
        String jdbcUsername = "root";
        String jdbcPassword = "Sjz@0626";
        String query = "SELECT user_id, activity, timestamp FROM user_logs";
        RowTypeInfo rowTypeInfo = getRowTypeInfo(jdbcUrl, jdbcDriver, jdbcUsername, jdbcPassword, query);

        JdbcInputFormat jdbcInputFormat = JdbcInputFormat.buildJdbcInputFormat()
                .setDrivername(jdbcDriver)
                .setDBUrl(jdbcUrl)
                .setUsername(jdbcUsername)
                .setPassword(jdbcPassword)
                .setQuery(query)
                .setRowTypeInfo(rowTypeInfo)
                .finish();


        DataStream<Row> jdbcStream = env.createInput(jdbcInputFormat);
        jdbcStream.print();
        SingleOutputStreamOperator<Demo.UserLog> map = jdbcStream.map(row -> {
            return new Demo.UserLog(row.getFieldAs(0), row.getFieldAs(1), row.getFieldAs(2));
        });
        map.print();

//        DataStream<Tuple3<Integer, String, Long>> userLogs = jdbcStream.map(row ->
//                new Tuple3<>(
//                        (Integer) row.getField(0),
//                        (String) row.getField(1),
//                        (Long) row.getField(2)
//                )
//        );
//
//        userLogs.print();

        env.execute("Flink Doris Read Demo");
    }


    private static RowTypeInfo getRowTypeInfo(String jdbcUrl, String jdbcDriver, String jdbcUsername, String jdbcPassword, String query) throws SQLException, ClassNotFoundException {
        Class.forName(jdbcDriver);
        try (Connection connection = DriverManager.getConnection(jdbcUrl, jdbcUsername, jdbcPassword);
             Statement statement = connection.createStatement();
             ResultSet resultSet = statement.executeQuery(query)) {

            ResultSetMetaData metaData = resultSet.getMetaData();
            int columnCount = metaData.getColumnCount();
            TypeInformation<?>[] types = new TypeInformation<?>[columnCount];

            for (int i = 0; i < columnCount; i++) {
                String columnTypeName = metaData.getColumnTypeName(i + 1);
                switch (columnTypeName) {
                    case "INT":
                    case "INTEGER":
                        types[i] = BasicTypeInfo.INT_TYPE_INFO;
                        break;
                    case "BIGINT":
                        types[i] = BasicTypeInfo.LONG_TYPE_INFO;
                        break;
                    case "VARCHAR":
                    case "CHAR":
                    case "TEXT":
                        types[i] = BasicTypeInfo.STRING_TYPE_INFO;
                        break;
                    default:
                        types[i] = BasicTypeInfo.STRING_TYPE_INFO; // default to STRING_TYPE_INFO for simplicity
                        break;
                }
            }
            return new RowTypeInfo(types);
        }
    }




}

